package oneTouchVideo

import (
	"context"
	"encoding/json"
	"fmt"
	"go-api/internal/dao/model"
	"go-api/internal/dao/model/mysql"
	"go-api/internal/dao/schema"
	"go-api/internal/helper"
	"go-api/internal/svc"
	"go-api/pkg/gpt"
	"gorm.io/gorm"
	"strings"
	"time"

	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/threading"
)

const (
	workerPoolSize       = 1000 // 并发工作协程数
	batchSize            = 500  // 批量处理数量
	maxRetries           = 3    // 最大重试次数
	lockExpiration       = 2 * time.Minute
	statusWordInit       = 0
	statusWordComplete   = 1
	statusGeneratingFail = 22
)

type WorkerPool struct {
	maxWorkers int
	taskChan   chan func()
}

func NewWorkerPool(maxWorkers int) *WorkerPool {
	pool := &WorkerPool{
		maxWorkers: maxWorkers,
		taskChan:   make(chan func(), 1e6), // 百万级缓冲
	}
	for i := 0; i < maxWorkers; i++ {
		go pool.worker()
	}
	return pool
}

func (p *WorkerPool) Submit(task func()) {
	p.taskChan <- task
}

func (p *WorkerPool) worker() {
	for task := range p.taskChan {
		task()
	}
}

type OneTouchVideoCron struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	db     *gorm.DB
	pool   *WorkerPool
}

func NewOneTouchVideoCron(ctx context.Context, svcCtx *svc.ServiceContext) *OneTouchVideoCron {
	return &OneTouchVideoCron{
		ctx:    ctx,
		svcCtx: svcCtx,
		db:     svcCtx.DbSelect.GetDb(ctx, model.DB),
		pool:   NewWorkerPool(workerPoolSize),
	}
}

// Register 注册定时任务，这里改为启动监听 Redis 队列的协程
func (c *OneTouchVideoCron) Register() {
	go c.RunWordSearchJobFromQueue() // 启动监听 Redis 队列的协程
	videoGenerateCron := NewVideoGenerateCron(c.ctx, c.svcCtx)
	go videoGenerateCron.RunVideoGenerateJob()
	videoResultCron := NewVideoResultCron(c.ctx, c.svcCtx)
	go videoResultCron.RunVideoResultJob()
}

// RunWordSearchJobFromQueue 从 Redis 队列中获取任务并处理
func (c *OneTouchVideoCron) RunWordSearchJobFromQueue() {
	for {
		// 从 Redis 队列中获取任务 ID
		taskIDStr, err := c.svcCtx.RedisClient.LPop(c.ctx, "laravel_database_video_word_search_queue").Result()
		if err != nil {
			time.Sleep(1 * time.Second) // 如果获取失败，等待 1 秒后重试
			continue
		}
		if taskIDStr == "" {
			time.Sleep(1 * time.Second) // 如果队列为空，等待 1 秒后重试
			continue
		}

		// 将任务 ID 转换为整数
		var taskID int64
		fmt.Sscanf(taskIDStr, "%d", &taskID)

		// 根据任务 ID 查询任务信息
		var tasks []*schema.CustomerOneTouchVideoTask
		err = c.db.Table("customer_one_touch_video_task_info").
			Select("customer_one_touch_video_task_info.*, customer_one_touch_video_task.*").
			Joins("JOIN customer_one_touch_video_task ON customer_one_touch_video_task_info.customer_one_touch_video_task_id = customer_one_touch_video_task.id").
			Where("customer_one_touch_video_task_info.customer_one_touch_video_task_id = ? AND customer_one_touch_video_task_info.status = ?", taskID, statusWordInit).
			Limit(batchSize).
			Find(&tasks).Error
		if err != nil {
			logx.Errorf("查询任务失败: %v", err)
			continue
		}
		if len(tasks) == 0 {
			continue
		}

		// 分发任务到 worker pool
		for _, task := range tasks {
			c.pool.Submit(func() {
				c.processWordSearchTask(task)
			})
		}
	}
}

func (c *OneTouchVideoCron) processWordSearchTask(task *schema.CustomerOneTouchVideoTask) {
	// 使用 Redis 分布式锁保证单个任务原子操作
	lockKey := fmt.Sprintf("video_task:%d:lock", task.ID)
	if !c.svcCtx.RedisClient.SetNX(c.ctx, lockKey, 1, lockExpiration).Val() {
		return
	}
	defer c.svcCtx.RedisClient.Del(c.ctx, lockKey)

	var content map[string]interface{} // 将 content 变量定义移到循环外部
	// 处理任务
	for retry := 0; retry < maxRetries; retry++ {
		var err error
		content, err = c.generateContent(task.Keyword, task.WebSiteId)
		if err != nil {
			logx.Errorf("生成内容失败: %v", err)
			time.Sleep(1 * time.Second << retry) // 指数退避
			continue
		}
		if content != nil && content["title"] != nil && content["text"] != nil && content["hashtag"] != nil {
			c.processContent(task, content, task.IsFilterSensitive)
			// 发布任务到视频生成队列
			payload, _ := json.Marshal(task)
			c.svcCtx.RedisClient.RPush(c.ctx, "video_generate_queue", payload)
			break
		}
		time.Sleep(1 * time.Second << retry) // 指数退避
	}

	if content == nil || content["title"] == nil || content["text"] == nil || content["hashtag"] == nil {
		infoModel := mysql.NewCustomerOneTouchVideoTaskInfoModel(c.ctx, c.svcCtx)
		_, err := infoModel.FindOne(task.ID)
		if err != nil {
			logx.Errorf("查找任务信息失败: %v", err)
			return
		}
		updateMap := map[string]interface{}{
			"status": statusGeneratingFail,
		}
		err = infoModel.UpdateByMap(task.ID, updateMap)
		if err != nil {
			logx.Errorf("更新任务状态失败: %v", err)
		}
	}
}

// generateContent 调用 AI 生成内容（带连接池和超时控制）
func (c *OneTouchVideoCron) generateContent(keyword string, websiteId int64) (map[string]interface{}, error) {
	searchWord := fmt.Sprintf("你作为%s的短视频运营人员，首先根据%s随机生成一个标题，再根据这个标题创作出80字以上的短视频宣传文案和随机3个数组标签话题。只返回json，参考格式:title:,text:,hashtag:", keyword, keyword)
	ws := gpt.NewWordSearch(searchWord, gpt.GPT_VERSION_FOUR)

	// 使用带超时的上下文
	ctx, cancel := context.WithTimeout(c.ctx, 30*time.Second)
	defer cancel()

	// 使用通道控制并发量
	resultChan := make(chan map[string]interface{}, 1)
	errChan := make(chan error, 1)
	threading.GoSafe(func() {
		res, err := ws.DouBao()
		if err != nil {
			errChan <- err
			return
		}
		resultChan <- res
	})

	select {
	case res := <-resultChan:
		return res, nil
	case err := <-errChan:
		return nil, err
	case <-ctx.Done():
		logx.Errorf("DouBao请求超时，关键词: %s", keyword)
		return nil, fmt.Errorf("DouBao请求超时，关键词: %s", keyword)
	}
}

// processContent 处理生成的内容并更新数据库
func (c *OneTouchVideoCron) processContent(task *schema.CustomerOneTouchVideoTask, content map[string]interface{}, isFilterSensitive int8) {
	// 开启事务
	tx := c.db.Begin()
	defer func() {
		if r := recover(); r != nil {
			tx.Rollback()
		}
	}()

	var title, text, tag string
	if isFilterSensitive == 1 {
		titleVal, titleExists := content["title"]
		textVal, textExists := content["text"]
		hashtagVal, hashtagExists := content["hashtag"]

		if titleExists {
			title = fmt.Sprintf("%v", titleVal)
		}
		if textExists {
			text = fmt.Sprintf("%v", textVal)
		}
		if hashtagExists {
			if hashtagArray, ok := hashtagVal.([]interface{}); ok {
				var tags []string
				for _, t := range hashtagArray {
					tags = append(tags, fmt.Sprintf("%v", t))
				}
				tag = strings.Join(tags, ",")
			}
		}

		// 创建敏感词过滤实例
		sensitiveHelper := helper.SensitiveHelper{}
		var err error
		// 过滤标题
		title, err = sensitiveHelper.FilterKeyword(isFilterSensitive == 1, title)
		if err != nil {
			tx.Rollback()
			logx.Errorf("过滤标题敏感词失败: %v", err)
			return
		}
		// 过滤文案
		text, err = sensitiveHelper.FilterKeyword(isFilterSensitive == 1, text)
		if err != nil {
			tx.Rollback()
			logx.Errorf("过滤文案敏感词失败: %v", err)
			return
		}
		// 过滤标签
		tag, err = sensitiveHelper.FilterKeyword(isFilterSensitive == 1, tag)
		if err != nil {
			tx.Rollback()
			logx.Errorf("过滤标签敏感词失败: %v", err)
			return
		}
	} else {
		if titleVal, ok := content["title"]; ok {
			title = fmt.Sprintf("%v", titleVal)
		}
		if textVal, ok := content["text"]; ok {
			text = fmt.Sprintf("%v", textVal)
		}
		if hashtagVal, ok := content["hashtag"]; ok {
			if hashtagArray, ok := hashtagVal.([]interface{}); ok {
				var tags []string
				for _, t := range hashtagArray {
					tags = append(tags, fmt.Sprintf("%v", t))
				}
				tag = strings.Join(tags, ",")
			}
		}
	}

	// 更新任务信息
	updateMap := map[string]interface{}{
		"title":    title,
		"captions": text,
		"tag":      tag,
		"status":   statusWordComplete,
	}

	if err := mysql.NewCustomerOneTouchVideoTaskInfoModel(c.ctx, c.svcCtx).UpdateByMap(task.ID, updateMap); err != nil {
		tx.Rollback()
		logx.Errorf("更新任务失败: %v", err)
		return
	}

	// 更新主任务状态
	if err := tx.Model(&schema.CustomerOneTouchVideoTask{}).
		Where("id =?", task.ID).
		Update("status", statusWordComplete).
		Error; err != nil {
		tx.Rollback()
		logx.Errorf("更新主任务状态失败: %v", err)
		return
	}

	tx.Commit()
}

// 公共方法
func (c *OneTouchVideoCron) acquireLock(lockKey string) bool {
	return c.svcCtx.RedisClient.SetNX(c.ctx,
		fmt.Sprintf("cron_lock:%s", lockKey),
		1,
		4*time.Minute).Val()
}
